#########################################################################
#
# Copyright (C) 2016 OSGeo
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
#########################################################################
import logging
from datetime import datetime, timedelta
from tastypie.test import ResourceTestCaseMixin, TestApiClient
from unittest.mock import patch
from urllib.parse import urlencode
from uuid import uuid4
import json
from django.conf import settings
from django.urls import reverse
from django.contrib.auth.models import Group
from django.contrib.auth import get_user_model
from django.test.utils import override_settings

from guardian.shortcuts import get_anonymous_user

from geonode import geoserver
from geonode.geoserver.manager import GeoServerResourceManager
from geonode.maps.models import Map
from geonode.layers.models import Dataset
from geonode.documents.models import Document
from geonode.base.models import (
    ExtraMetadata,
    Thesaurus,
    ThesaurusLabel,
    ThesaurusKeyword,
    ThesaurusKeywordLabel,
    ResourceBase,
)
from geonode.utils import check_ogc_backend
from geonode.decorators import on_ogc_backend
from geonode.groups.models import GroupProfile
from geonode.base.auth import get_or_create_token
from geonode.tests.base import GeoNodeBaseTestSupport
from geonode.base.populate_test_data import all_public, create_models, remove_models
from geonode.security.registry import permissions_registry
from geonode.assets.models import Asset
from django.core.files.uploadedfile import SimpleUploadedFile
from geonode.base.models import Link
from django.contrib.auth.models import Permission
from django.contrib.contenttypes.models import ContentType
from geonode.upload.models import UploadSizeLimit


logger = logging.getLogger(__name__)


class UserAndTokenInfoApiTests(GeoNodeBaseTestSupport):
    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        create_models(type=cls.get_type, integration=cls.get_integration)
        all_public()

    @classmethod
    def tearDownClass(cls):
        super().tearDownClass()
        remove_models(cls.get_obj_ids, type=cls.get_type, integration=cls.get_integration)

    def test_userinfo_response(self):
        userinfo_url = reverse("userinfo")
        _user = get_user_model().objects.get(username="bobby")
        self.client.login(username="bobby", password="bob")
        response = self.client.get(userinfo_url)
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json()["sub"], str(_user.pk))
        self.client.logout()
        response = self.client.get(userinfo_url)
        self.assertEqual(response.status_code, 401)

    def test_tokeninfo_response(self):
        tokeninfo_url = reverse("tokeninfo")
        _user = get_user_model().objects.get(username="bobby")
        token = get_or_create_token(_user)
        response = self.client.post(tokeninfo_url, data={"token": token})
        self.assertEqual(response.status_code, 200)
        response_json = response.json()
        self.assertEqual(response_json["access_token"], token.token)
        self.assertEqual(response_json["user_id"], _user.pk)


class PermissionsApiTests(ResourceTestCaseMixin, GeoNodeBaseTestSupport):
    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        create_models(type=cls.get_type, integration=cls.get_integration)
        all_public()

    @classmethod
    def tearDownClass(cls):
        super().tearDownClass()
        remove_models(cls.get_obj_ids, type=cls.get_type, integration=cls.get_integration)

    def setUp(self):
        super().setUp()
        self.user = "admin"
        self.passwd = "admin"
        self.perm_spec = {"users": {}, "groups": {}}

    def test_dataset_get_list_unauth_all_public(self):
        """
        Test that the correct number of layers are returned when the
        client is not logged in and all are public
        """
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "datasets"})
        resp = self.api_client.get(list_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 8)

    def test_datasets_get_list_unauth_some_public(self):
        """
        Test that if a layer is not public then not all are returned when the
        client is not logged in
        """
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "datasets"})

        layer = Dataset.objects.first()
        layer.set_permissions(self.perm_spec)

        resp = self.api_client.get(list_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 7)

    def test_datasets_get_list_auth_some_public(self):
        """
        Test that if a layer is not public then all are returned if the
        client is not logged in
        """
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "datasets"})

        self.api_client.client.login(username=self.user, password=self.passwd)
        layer = Dataset.objects.first()
        layer.set_permissions(self.perm_spec)

        resp = self.api_client.get(list_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 8)

    def test_dataset_get_list_dataset_private_to_one_user(self):
        """
        Test that if a layer is only visible by admin, then does not appear
        in the unauthenticated list nor in the list when logged is as bobby
        """
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "datasets"})

        perm_spec = {"users": {"admin": ["view_resourcebase"]}, "groups": {}}
        layer = Dataset.objects.first()
        layer.set_permissions(perm_spec)
        resp = self.api_client.get(list_url)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 7)

        self.api_client.client.login(username="bobby", password="bob")
        resp = self.api_client.get(list_url)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 7)

        self.api_client.client.login(username=self.user, password=self.passwd)
        resp = self.api_client.get(list_url)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 8)

        layer.is_published = False
        layer.save()

        # with resource publishing
        with self.settings(RESOURCE_PUBLISHING=True):
            resp = self.api_client.get(list_url)
            self.assertGreaterEqual(len(self.deserialize(resp)["objects"]), 7)

            self.api_client.client.login(username="bobby", password="bob")
            resp = self.api_client.get(list_url)
            self.assertGreaterEqual(len(self.deserialize(resp)["objects"]), 7)

            self.api_client.client.login(username=self.user, password=self.passwd)
            resp = self.api_client.get(list_url)
            self.assertGreaterEqual(len(self.deserialize(resp)["objects"]), 7)

    def test_dataset_get_detail_unauth_dataset_not_public(self):
        """
        Test that layer detail gives 404 when not public and not logged in
        """
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "datasets"})

        resp = self.client.get(list_url)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 8)

        layer = Dataset.objects.first()
        layer.set_permissions(self.perm_spec)
        layer.clear_dirty_state()

        resp = self.client.get(list_url)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 7)
        self.assertHttpNotFound(self.client.get(f"{list_url + str(layer.id)}/"))

        self.client.login(username=self.user, password=self.passwd)
        self.assertValidJSONResponse(self.client.get(f"{list_url + str(layer.id)}/"))

        # with delayed security
        with self.settings(DELAYED_SECURITY_SIGNALS=True, GEOFENCE_SECURITY_ENABLED=True):
            if check_ogc_backend(geoserver.BACKEND_PACKAGE):
                gm = GeoServerResourceManager()
                gm.set_permissions(layer.uuid, instance=layer, permissions=self.perm_spec)
                self.assertTrue(layer.dirty_state)

                self.client.login(username=self.user, password=self.passwd)
                resp = self.client.get(list_url)
                self.assertEqual(len(self.deserialize(resp)["objects"]), 7)  # admin can't see resources in dirty_state

                self.client.logout()
                resp = self.client.get(list_url)
                self.assertEqual(len(self.deserialize(resp)["objects"]), 7)

                from django.contrib.auth import get_user_model

                get_user_model().objects.create(
                    username="imnew",
                    password="pbkdf2_sha256$12000$UE4gAxckVj4Z$N6NbOXIQWWblfInIoq/Ta34FdRiPhawCIZ+sOO3YQs=",
                )
                self.client.login(username="imnew", password="thepwd")
                resp = self.client.get(list_url)
                self.assertEqual(len(self.deserialize(resp)["objects"]), 7)

    def test_new_user_has_access_to_old_datasets(self):
        """Test that a new user can access the public available layers"""
        from django.contrib.auth import get_user_model

        get_user_model().objects.create(
            username="imnew",
            password="pbkdf2_sha256$12000$UE4gAxckVj4Z$N\
            6NbOXIQWWblfInIoq/Ta34FdRiPhawCIZ+sOO3YQs=",
        )

        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "datasets"})

        self.api_client.client.login(username="imnew", password="thepwd")
        resp = self.api_client.get(list_url)
        self.assertValidJSONResponse(resp)
        self.assertGreaterEqual(len(self.deserialize(resp)["objects"]), 7)

        # with delayed security
        if check_ogc_backend(geoserver.BACKEND_PACKAGE):
            _ogc_geofence_enabled = settings.OGC_SERVER
            try:
                _ogc_geofence_enabled["default"]["GEOFENCE_SECURITY_ENABLED"] = True
                with self.settings(
                    DELAYED_SECURITY_SIGNALS=True,
                    OGC_SERVER=_ogc_geofence_enabled,
                    DEFAULT_ANONYMOUS_VIEW_PERMISSION=True,
                ):
                    layer = Dataset.objects.first()
                    layer.set_default_permissions()
                    layer.refresh_from_db()
                    # self.assertTrue(layer.dirty_state)

                    self.client.login(username=self.user, password=self.passwd)
                    resp = self.client.get(list_url)
                    self.assertGreaterEqual(len(self.deserialize(resp)["objects"]), 7)

                    self.client.logout()
                    resp = self.client.get(list_url)
                    self.assertGreaterEqual(len(self.deserialize(resp)["objects"]), 7)

                    self.client.login(username="imnew", password="thepwd")
                    resp = self.client.get(list_url)
                    self.assertGreaterEqual(len(self.deserialize(resp)["objects"]), 7)
            finally:
                _ogc_geofence_enabled["default"]["GEOFENCE_SECURITY_ENABLED"] = False

    @on_ogc_backend(geoserver.BACKEND_PACKAGE)
    def test_outh_token(self):
        user = "admin"
        _user = get_user_model().objects.get(username=user)
        token = get_or_create_token(_user)
        auth_header = f"Bearer {token}"
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "datasets"})

        with self.settings(SESSION_EXPIRED_CONTROL_ENABLED=False, DELAYED_SECURITY_SIGNALS=False):
            # all public
            resp = self.api_client.get(list_url)
            self.assertValidJSONResponse(resp)
            self.assertGreaterEqual(len(self.deserialize(resp)["objects"]), 7)

            perm_spec = {"users": {"admin": ["view_resourcebase"]}, "groups": {}}
            layer = Dataset.objects.first()
            layer.set_permissions(perm_spec)
            resp = self.api_client.get(list_url)
            self.assertGreaterEqual(len(self.deserialize(resp)["objects"]), 7)

            resp = self.api_client.get(list_url, authentication=auth_header)
            self.assertGreaterEqual(len(self.deserialize(resp)["objects"]), 7)

            layer.is_published = False
            layer.save()

    @override_settings(API_LOCKDOWN=True)
    def test_api_lockdown_false(self):
        profiles_list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "profiles"})

        # test if results are returned for anonymous users if API_LOCKDOWN is set to False in settings
        filter_url = profiles_list_url

        with self.settings(API_LOCKDOWN=False):
            # anonymous
            resp = self.api_client.get(filter_url)
            self.assertValidJSONResponse(resp)
            self.assertEqual(len(self.deserialize(resp)["objects"]), 0)
            # admin
            self.api_client.client.login(username="admin", password="admin")
            resp = self.api_client.get(filter_url)
            self.assertValidJSONResponse(resp)
            self.assertEqual(len(self.deserialize(resp)["objects"]), 9)

    @override_settings(API_LOCKDOWN=True)
    def test_profiles_lockdown(self):
        profiles_list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "profiles"})

        filter_url = profiles_list_url
        resp = self.api_client.get(filter_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 0)

        # now test with logged in user
        self.api_client.client.login(username="bobby", password="bob")
        resp = self.api_client.get(filter_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 6)
        # Returns limited info about other users
        bobby = get_user_model().objects.get(username="bobby")
        profiles = self.deserialize(resp)["objects"]
        for profile in profiles:
            if profile["username"] == "bobby":
                self.assertEqual(profile.get("email"), bobby.email)
            else:
                self.assertIsNone(profile.get("email"))

    @override_settings(API_LOCKDOWN=True)
    def test_owners_lockdown(self):
        owners_list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "owners"})

        filter_url = owners_list_url

        resp = self.api_client.get(filter_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 0)

        # now test with logged in user
        self.api_client.client.login(username="bobby", password="bob")
        resp = self.api_client.get(filter_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 6)
        # Returns limited info about other users
        bobby = get_user_model().objects.get(username="bobby")
        owners = self.deserialize(resp)["objects"]
        for owner in owners:
            if owner["username"] == "bobby":
                self.assertEqual(owner.get("email"), bobby.email)
            else:
                self.assertIsNone(owner.get("email"))
                self.assertIsNone(owner.get("first_name"))

        # now test with logged in admin
        self.api_client.client.login(username="admin", password="admin")
        resp = self.api_client.get(filter_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 9)

    @override_settings(API_LOCKDOWN=True)
    def test_groups_lockdown(self):
        groups_list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "groups"})

        filter_url = groups_list_url

        resp = self.api_client.get(filter_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 0)

        # now test with logged in user
        self.api_client.client.login(username="bobby", password="bob")
        resp = self.api_client.get(filter_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 1)

    @override_settings(API_LOCKDOWN=True)
    def test_regions_lockdown(self):
        region_list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "regions"})

        filter_url = region_list_url

        resp = self.api_client.get(filter_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 0)

        self.api_client.client.login(username="bobby", password="bob")
        resp = self.api_client.get(filter_url)
        self.assertValidJSONResponse(resp)
        self.assertTrue(len(self.deserialize(resp)["objects"]) >= 200)

    @override_settings(API_LOCKDOWN=True)
    def test_tags_lockdown(self):
        tag_list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "keywords"})

        filter_url = tag_list_url

        resp = self.api_client.get(filter_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 0)

        self.api_client.client.login(username="bobby", password="bob")
        resp = self.api_client.get(filter_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 5)


class SearchApiTests(ResourceTestCaseMixin, GeoNodeBaseTestSupport):
    """Test the search"""

    #  loading test thesausuri and initial data
    fixtures = ["initial_data.json", "group_test_data.json", "default_oauth_apps.json", "test_thesaurus.json"]

    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        create_models(type=cls.get_type, integration=cls.get_integration)
        all_public()

    @classmethod
    def tearDownClass(cls):
        super().tearDownClass()
        remove_models(cls.get_obj_ids, type=cls.get_type, integration=cls.get_integration)

    def setUp(self):
        super().setUp()

        self.norman = get_user_model().objects.get(username="norman")
        self.norman.groups.add(Group.objects.get(name="anonymous"))
        self.test_user = get_user_model().objects.get(username="test_user")
        self.test_user.groups.add(Group.objects.get(name="anonymous"))
        self.bar = GroupProfile.objects.get(slug="bar")
        self.anonymous_user = get_anonymous_user()
        self.profiles_list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "profiles"})
        self.groups_list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "groups"})

    def test_profiles_filters(self):
        """Test profiles filtering"""

        with self.settings(API_LOCKDOWN=False):
            filter_url = self.profiles_list_url

            resp = self.api_client.get(filter_url)
            self.assertValidJSONResponse(resp)
            self.assertEqual(len(self.deserialize(resp)["objects"]), 0)

            filter_url = f"{self.profiles_list_url}?name__icontains=norm"
            # Anonymous
            resp = self.api_client.get(filter_url)
            self.assertValidJSONResponse(resp)
            self.assertEqual(len(self.deserialize(resp)["objects"]), 0)

            self.api_client.client.login(username="admin", password="admin")
            resp = self.api_client.get(filter_url)
            self.assertValidJSONResponse(resp)
            self.assertEqual(len(self.deserialize(resp)["objects"]), 1)

            filter_url = f"{self.profiles_list_url}?name__icontains=NoRmAN"

            resp = self.api_client.get(filter_url)
            self.assertValidJSONResponse(resp)
            self.assertEqual(len(self.deserialize(resp)["objects"]), 1)

            filter_url = f"{self.profiles_list_url}?name__icontains=bar"

            resp = self.api_client.get(filter_url)
            self.assertValidJSONResponse(resp)
            self.assertEqual(len(self.deserialize(resp)["objects"]), 0)

    def test_groups_filters(self):
        """Test groups filtering"""

        with self.settings(API_LOCKDOWN=False):
            filter_url = self.groups_list_url

            resp = self.api_client.get(filter_url)
            self.assertValidJSONResponse(resp)
            self.assertEqual(len(self.deserialize(resp)["objects"]), 1)

            filter_url = f"{self.groups_list_url}?name__icontains=bar"

            resp = self.api_client.get(filter_url)
            self.assertValidJSONResponse(resp)
            self.assertEqual(len(self.deserialize(resp)["objects"]), 1)

            filter_url = f"{self.groups_list_url}?name__icontains=BaR"

            resp = self.api_client.get(filter_url)
            self.assertValidJSONResponse(resp)
            self.assertEqual(len(self.deserialize(resp)["objects"]), 1)

            filter_url = f"{self.groups_list_url}?name__icontains=foo"

            resp = self.api_client.get(filter_url)
            self.assertValidJSONResponse(resp)
            self.assertEqual(len(self.deserialize(resp)["objects"]), 0)

    def test_category_filters(self):
        """Test category filtering"""
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "datasets"})

        # check we get the correct layers number returnered filtering on one
        # and then two different categories
        filter_url = f"{list_url}?category__identifier=location"

        resp = self.api_client.get(filter_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 3)

        filter_url = f"{list_url}?category__identifier__in=location&category__identifier__in=biota"

        resp = self.api_client.get(filter_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 5)

    def test_metadata_filters(self):
        """Test category filtering"""
        _r = Dataset.objects.first()
        _m = ExtraMetadata.objects.create(
            resource=_r,
            metadata={
                "name": "metadata-updated",
                "slug": "metadata-slug-updated",
                "help_text": "this is the help text-updated",
                "field_type": "str-updated",
                "value": "my value-updated",
                "category": "category",
            },
        )

        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "datasets"})
        _r.metadata.add(_m)
        # check we get the correct layers number returnered filtering on one
        # and then two different categories
        filter_url = f"{list_url}?metadata__category=category"

        resp = self.api_client.get(filter_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 1)

        filter_url = f"{list_url}?metadata__category=not-existing-category"

        resp = self.api_client.get(filter_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 0)

    def test_tag_filters(self):
        """Test keywords filtering"""
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "datasets"})

        # check we get the correct layers number returnered filtering on one
        # and then two different keywords
        filter_url = f"{list_url}?keywords__slug=layertagunique"

        resp = self.api_client.get(filter_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 1)

        filter_url = f"{list_url}?keywords__slug__in=layertagunique&keywords__slug__in=populartag"

        resp = self.api_client.get(filter_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 8)

    def test_owner_filters(self):
        """Test owner filtering"""
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "datasets"})

        # check we get the correct layers number returnered filtering on one
        # and then two different owners
        filter_url = f"{list_url}?owner__username=user1"

        resp = self.api_client.get(filter_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 1)

        filter_url = f"{list_url}?owner__username__in=user1&owner__username__in=foo"

        resp = self.api_client.get(filter_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 2)

    def test_title_filter(self):
        """Test title filtering"""
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "datasets"})

        # check we get the correct layers number returnered filtering on the
        # title
        filter_url = f"{list_url}?title=layer2"

        resp = self.api_client.get(filter_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 1)

    def test_date_filter(self):
        """Test date filtering"""
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "datasets"})

        # check we get the correct layers number returnered filtering on the
        # dates
        step = timedelta(days=60)
        now = datetime.now()
        fstring = "%Y-%m-%d"

        def to_date(val):
            return val.date().strftime(fstring)

        d1 = to_date(now - step)
        filter_url = f"{list_url}?date__exact={d1}"

        resp = self.api_client.get(filter_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 0)

        d3 = to_date(now - (3 * step))
        filter_url = f"{list_url}?date__gte={d3}"

        resp = self.api_client.get(filter_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 3)

        d4 = to_date(now - (4 * step))
        filter_url = f"{list_url}?date__range={d4},{to_date(now)}"

        resp = self.api_client.get(filter_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 4)

    def test_extended_text_filter(self):
        """Test that the extended text filter works as expected"""
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "datasets"})

        filter_url = (
            f"{list_url}?title__icontains=layer2&abstract__icontains=layer2&purpose__icontains=layer2&f_method=or"
        )

        resp = self.api_client.get(filter_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(len(self.deserialize(resp)["objects"]), 1)

    def test_the_api_should_return_all_datasets_with_metadata_false(self):
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "datasets"})
        user = get_user_model().objects.get(username="admin")
        token = get_or_create_token(user)
        auth_header = f"Bearer {token}"

        resp = self.api_client.get(list_url, authentication=auth_header)
        self.assertValidJSONResponse(resp)
        self.assertEqual(Dataset.objects.filter(metadata_only=False).count(), resp.json()["meta"]["total_count"])

    def test_the_api_should_return_all_datasets_with_metadata_true(self):
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "datasets"})
        user = get_user_model().objects.get(username="admin")
        token = get_or_create_token(user)
        auth_header = f"Bearer {token}"

        url = f"{list_url}?metadata_only=True"
        resp = self.api_client.get(url, authentication=auth_header)
        self.assertValidJSONResponse(resp)
        self.assertEqual(Dataset.objects.filter(metadata_only=True).count(), resp.json()["meta"]["total_count"])

    def test_the_api_should_return_all_documents_with_metadata_false(self):
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "documents"})

        resp = self.api_client.get(list_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(Document.objects.filter(metadata_only=False).count(), resp.json()["meta"]["total_count"])

    def test_the_api_should_return_all_documents_with_metadata_true(self):
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "documents"})

        url = f"{list_url}?metadata_only=True"
        resp = self.api_client.get(url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(Document.objects.filter(metadata_only=True).count(), resp.json()["meta"]["total_count"])

    def test_the_api_should_return_all_maps_with_metadata_false(self):
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "maps"})

        resp = self.api_client.get(list_url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(Map.objects.filter(metadata_only=False).count(), resp.json()["meta"]["total_count"])

    def test_the_api_should_return_all_maps_with_metadata_true(self):
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "maps"})

        url = f"{list_url}?metadata_only=True"
        resp = self.api_client.get(url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(Map.objects.filter(metadata_only=True).count(), resp.json()["meta"]["total_count"])

    def test_api_will_return_a_valid_json_response(self):
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "thesaurus/keywords"})

        resp = self.api_client.get(list_url)
        self.assertValidJSONResponse(resp)

    def test_will_return_empty_if_the_thesaurus_does_not_exists(self):
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "thesaurus/keywords"})

        url = f"{list_url}?thesaurus=invalid-identifier"
        resp = self.api_client.get(url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(resp.json()["meta"]["total_count"], 0)

    def test_will_return_keywords_for_the_selected_thesaurus_if_exists(self):
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "thesaurus/keywords"})

        url = f"{list_url}?thesaurus=inspire-theme"
        resp = self.api_client.get(url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(resp.json()["meta"]["total_count"], 36)

    def test_will_return_empty_if_the_alt_label_does_not_exists(self):
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "thesaurus/keywords"})

        url = f"{list_url}?alt_label=invalid-alt_label"
        resp = self.api_client.get(url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(resp.json()["meta"]["total_count"], 0)

    def test_will_return_keywords_for_the_selected_alt_label_if_exists(self):
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "thesaurus/keywords"})

        url = f"{list_url}?alt_label=ac"
        resp = self.api_client.get(url)
        self.assertValidJSONResponse(resp)
        self.assertEqual(resp.json()["meta"]["total_count"], 1)

    def test_will_return_empty_if_the_kaywordId_does_not_exists(self):
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "thesaurus/keywords"})

        url = f"{list_url}?id=12365478954862"
        resp = self.api_client.get(url)
        print(self.deserialize(resp))
        self.assertValidJSONResponse(resp)
        self.assertEqual(resp.json()["meta"]["total_count"], 0)

    @patch("geonode.api.api.get_language")
    def test_will_return_expected_keyword_label_for_existing_lang(self, lang):
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "thesaurus/keywords"})

        lang.return_value = "de"
        url = f"{list_url}?thesaurus=inspire-theme"
        resp = self.api_client.get(url)
        # the german translations exists, for the other labels, the alt_label will be used
        expected_labels = [
            "",
            "ac",
            "Adressen",
            "af",
            "am",
            "au",
            "br",
            "bu",
            "cp",
            "ef",
            "el",
            "er",
            "foo_keyword",
            "ge",
            "gg",
            "gn",
            "hb",
            "hh",
            "hy",
            "lc",
            "lu",
            "mf",
            "mr",
            "nz",
            "of",
            "oi",
            "pd",
            "pf",
            "ps",
            "rs",
            "sd",
            "so",
            "sr",
            "su",
            "tn",
            "us",
        ]
        actual_labels = [x["alt_label"] for x in self.deserialize(resp)["objects"]]
        self.assertValidJSONResponse(resp)
        self.assertListEqual(expected_labels, actual_labels)

    @patch("geonode.api.api.get_language")
    def test_will_return_default_keyword_label_for_not_existing_lang(self, lang):
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "thesaurus/keywords"})

        lang.return_value = "ke"
        url = f"{list_url}?thesaurus=inspire-theme"
        resp = self.api_client.get(url)
        # no translations exists, the alt_label will be used for all keywords
        expected_labels = [
            "",
            "ac",
            "ad",
            "af",
            "am",
            "au",
            "br",
            "bu",
            "cp",
            "ef",
            "el",
            "er",
            "foo_keyword",
            "ge",
            "gg",
            "gn",
            "hb",
            "hh",
            "hy",
            "lc",
            "lu",
            "mf",
            "mr",
            "nz",
            "of",
            "oi",
            "pd",
            "pf",
            "ps",
            "rs",
            "sd",
            "so",
            "sr",
            "su",
            "tn",
            "us",
        ]
        actual_labels = [x["alt_label"] for x in self.deserialize(resp)["objects"]]
        self.assertValidJSONResponse(resp)
        self.assertListEqual(expected_labels, actual_labels)

    def test_the_api_should_return_all_map_categories_with_metadata_false(self):
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "categories"})

        url = f"{list_url}?type=map"
        resp = self.api_client.get(url)
        self.assertValidJSONResponse(resp)
        actual = sum([x["count"] for x in resp.json()["objects"]])
        self.assertEqual(9, actual)

    def test_the_api_should_return_all_map_categories_with_metadata_true(self):
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "categories"})

        x = Map.objects.get(title="map metadata true")
        x.metadata_only = False
        x.save()
        url = f"{list_url}?type=map"
        resp = self.api_client.get(url)
        self.assertValidJSONResponse(resp)
        # by adding a new layer, the total should increase
        actual = sum([x["count"] for x in resp.json()["objects"]])
        self.assertEqual(10, actual)

    def test_the_api_should_return_all_document_categories_with_metadata_false(self):
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "categories"})

        url = f"{list_url}?type=document"
        resp = self.api_client.get(url)
        self.assertValidJSONResponse(resp)
        actual = sum([x["count"] for x in resp.json()["objects"]])
        self.assertEqual(0, actual)

    def test_the_api_should_return_all_document_categories_with_metadata_true(self):
        list_url = reverse("api_dispatch_list", kwargs={"api_name": "api", "resource_name": "categories"})

        x = Document.objects.get(title="doc metadata true")
        x.metadata_only = False
        x.save()
        url = f"{list_url}?type=document"
        resp = self.api_client.get(url)
        self.assertValidJSONResponse(resp)
        # by adding a new layer, the total should increase
        actual = sum([x["count"] for x in resp.json()["objects"]])
        self.assertEqual(0, actual)


class ThesauriApiTests(GeoNodeBaseTestSupport):
    @classmethod
    def setUpClass(cls):
        super().setUpClass()

        cls.user = get_user_model().objects.create(username="user_00")
        cls.admin = get_user_model().objects.get(username="admin")

        cls._create_thesauri()
        cls._create_resources()

    @classmethod
    def tearDownClass(cls):
        super().tearDownClass()
        # remove_models(cls.get_obj_ids, type=cls.get_type, integration=cls.get_integration)

    def setUp(self):
        super().setUp()

        self.api_client = TestApiClient()

        self.assertEqual(self.admin.username, "admin")
        self.assertEqual(self.admin.is_superuser, True)

    @classmethod
    def _create_thesauri(cls):
        cls.thesauri = {}
        cls.thesauri_k = {}

        for tn in range(2):
            t = Thesaurus.objects.create(identifier=f"t_{tn}", title=f"Thesaurus {tn}")
            cls.thesauri[tn] = t
            for tl in (
                "en",
                "it",
            ):
                ThesaurusLabel.objects.create(thesaurus=t, lang=tl, label=f"TLabel {tn} {tl}")

            for tkn in range(10):
                tk = ThesaurusKeyword.objects.create(thesaurus=t, alt_label=f"alt_tkn{tkn}_t{tn}")
                cls.thesauri_k[f"{tn}_{tkn}"] = tk
                for tkl in (
                    "en",
                    "it",
                ):
                    ThesaurusKeywordLabel.objects.create(keyword=tk, lang=tkl, label=f"T{tn}_K{tkn}_{tkl}")

    @classmethod
    def _create_resources(self):
        public_perm_spec = {"users": {"AnonymousUser": ["view_resourcebase"]}, "groups": {}}

        for x in range(20):
            d: ResourceBase = ResourceBase.objects.create(
                title=f"dataset_{x:02}",
                uuid=str(uuid4()),
                owner=self.user,
                abstract=f"Abstract for dataset {x:02}",
                subtype="vector",
                is_approved=True,
                is_published=True,
            )

            # These are the assigned keywords to the Resources

            # RB00 ->            T1K0
            # RB01 ->  T0K0      T1K0
            # RB02 ->            T1K0
            # RB03 ->  T0K0      T1K0
            # RB04 ->            T1K0
            # RB05 ->  T0K0      T1K0
            # RB06 ->            T1K0
            # RB07 ->  T0K0      T1K0
            # RB08 ->            T1K0 T1K1
            # RB09 ->  T0K0      T1K0 T1K1
            # RB10 ->                 T1K1
            # RB11 ->  T0K0 T0K1      T1K1
            # RB12 ->                 T1K1
            # RB13 ->  T0K0 T0K1
            # RB14 ->
            # RB15 ->  T0K0 T0K1
            # RB16 ->
            # RB17 ->  T0K0 T0K1
            # RB18 ->
            # RB19 ->  T0K0 T0K1

            if x % 2 == 1:
                print(f"ADDING KEYWORDS {self.thesauri_k['0_0']} to RB {d}")
                d.tkeywords.add(self.thesauri_k["0_0"])
                d.save()
            if x % 2 == 1 and x > 10:
                print(f"ADDING KEYWORDS {self.thesauri_k['0_1']} to RB {d}")
                d.tkeywords.add(self.thesauri_k["0_1"])
                d.save()
            if x < 10:
                print(f"ADDING KEYWORDS {self.thesauri_k['1_0']} to RB {d}")
                d.tkeywords.add(self.thesauri_k["1_0"])
                d.save()
            if 7 < x < 13:
                d.tkeywords.add(self.thesauri_k["1_1"])
                d.save()

            d.set_permissions(public_perm_spec)

    def test_resources_filtered(self):
        # list_url = reverse("base-resources", kwargs={"api_name": "api", "resource_name": "base"})
        list_url = reverse("base-resources-list")

        for tks, exp, exp_and in (
            # single filter
            (("0_0",), 10, 10),
            (("0_1",), 5, 5),
            (("1_0",), 10, 10),
            (("1_1",), 5, 5),
            # same thesaurus: OR
            (("0_0", "0_1"), 10, 5),
            (("1_0", "1_1"), 13, 2),
            # different thesauri: AND
            (("0_0", "1_0"), 5, 5),
            (("0_1", "1_0"), 0, 0),
            (("0_1", "1_0", "1_1"), 1, 0),
            (("0_0", "0_1", "1_0", "1_1"), 6, 0),
        ):
            logger.debug(f"Testing filters for {tks}")
            filter = [("filter{tkeywords}", self.thesauri_k[tk].id) for tk in tks]
            url = f"{list_url}?{urlencode(filter)}"
            resp = self.api_client.get(url).json()
            self.assertEqual(exp, resp["total"], f"Unexpected number of resources for default filter {tks}")

            filter = [("filter{tkeywords}", self.thesauri_k[tk].id) for tk in tks]
            url = f"{list_url}?{urlencode(filter + [('force_and', True)])}"
            resp = self.api_client.get(url).json()
            self.assertEqual(exp_and, resp["total"], f"Unexpected number of resources for FORCE_AND filter {tks}")


class OwnershipTransferTest(ResourceTestCaseMixin, GeoNodeBaseTestSupport):
    def setUp(self):
        super().setUp()
        self.user1 = get_user_model().objects.create_user(username="user1", password="password1")
        self.user2 = get_user_model().objects.create_user(username="user2", password="password2")
        self.user3 = get_user_model().objects.create_user(username="user3", password="password3")
        self.client.login(username="user1", password="password1")

    def _get_updated_metadata(self, resource, old_owner, new_owner):
        return {
            "date_type": "publication",
            "hkeywords": [],
            "license": {"id": "not_specified", "label": "Not Specified"},
            "regions": [],
            "supplemental_information": "No information provided",
            "linkedresources": [],
            "contacts": {
                "owner": {"id": str(new_owner.id), "label": new_owner.username},
                "author": [{"id": str(old_owner.id), "label": old_owner.username}],
                "pointOfContact": [{"id": str(old_owner.id), "label": old_owner.username}],
            },
            "uuid": str(resource.uuid),
            "title": "test5",
            "abstract": "sasa",
            "date": "2025-08-13T15:30:18.724722+00:00",
            "tkeywords": {},
            "language": "eng",
        }

    def test_owner_change_permission_transfer(self):
        map_data = {"title": "Test Map", "abstract": "Test abstract", "owner": self.user1}
        map_resource = Map.objects.create(**map_data)
        map_resource.set_permissions({"users": {self.user1.username: ["change_resourcebase_metadata"]}})
        perms_user_1 = permissions_registry.get_perms(instance=map_resource, user=self.user1)
        self.assertIn("change_resourcebase_metadata", perms_user_1)
        metadata_url = f"/api/v2/metadata/instance/{map_resource.pk}/"

        updated_metadata = self._get_updated_metadata(map_resource, self.user1, self.user2)

        response = self.client.put(metadata_url, data=json.dumps(updated_metadata), content_type="application/json")
        self.assertEqual(response.status_code, 200)
        map_resource.refresh_from_db()

        perms_user_2 = permissions_registry.get_perms(instance=map_resource, user=self.user2)
        self.assertIn("change_resourcebase_metadata", perms_user_2)
        perms_user_1 = permissions_registry.get_perms(instance=map_resource, user=self.user1)
        self.assertNotIn("change_resourcebase_metadata", perms_user_1)

    def test_owner_change_does_not_affect_other_users_permissions(self):
        map_data = {"title": "Test Map", "abstract": "Test abstract", "owner": self.user1}
        map_resource = Map.objects.create(**map_data)
        map_resource.set_permissions(
            {
                "users": {
                    self.user1.username: ["change_resourcebase_metadata"],
                    self.user3.username: ["change_resourcebase_metadata"],
                }
            }
        )

        perms_user_1 = permissions_registry.get_perms(instance=map_resource, user=self.user1)
        self.assertIn("change_resourcebase_metadata", perms_user_1)
        perms_user_3 = permissions_registry.get_perms(instance=map_resource, user=self.user3)
        self.assertIn("change_resourcebase_metadata", perms_user_3)

        metadata_url = f"/api/v2/metadata/instance/{map_resource.pk}/"
        updated_metadata = self._get_updated_metadata(map_resource, self.user1, self.user2)

        response = self.client.put(metadata_url, data=json.dumps(updated_metadata), content_type="application/json")
        self.assertEqual(response.status_code, 200)
        map_resource.refresh_from_db()

        perms_user_1_after = permissions_registry.get_perms(instance=map_resource, user=self.user1)
        self.assertNotIn("change_resourcebase_metadata", perms_user_1_after)

        perms_user_2_after = permissions_registry.get_perms(instance=map_resource, user=self.user2)
        self.assertIn("change_resourcebase_metadata", perms_user_2_after)

        perms_user_3_after = permissions_registry.get_perms(instance=map_resource, user=self.user3)
        self.assertIn("change_resourcebase_metadata", perms_user_3_after)


class AssetApiTests(GeoNodeBaseTestSupport):
    def setUp(self):
        super().setUp()
        self.admin_user = get_user_model().objects.get(username="admin")
        self.test_user = get_user_model().objects.create_user(
            username="test_user12", email="testuser@example.com", password="testpass123"
        )
        self.resource = ResourceBase.objects.create(
            title="Test Resource for Assets",
            owner=self.admin_user,
            uuid=str(uuid4()),
        )
        content_type = ContentType.objects.get_for_model(ResourceBase)
        permission = Permission.objects.get(
            codename="add_resourcebase",
            content_type=content_type,
        )
        self.test_user.user_permissions.add(permission)
        self.resource.set_permissions(
            {"users": {self.test_user.username: ["view_resourcebase", "change_resourcebase"]}}
        )

    def _create_dummy_file(self, filename="test_file.txt", content=b"test content"):
        return SimpleUploadedFile(filename, content, content_type="text/plain")

    def test_asset_creation_and_link(self):
        """
        Tests that a user with correct permissions can upload an asset
        and that it is correctly linked to the specified resource.
        """
        self.client.force_login(self.test_user)
        url = reverse("base-resources-assets", kwargs={"pk": self.resource.pk})
        file = self._create_dummy_file()
        data = {"files": file, "title": "My Linked Asset"}

        initial_asset_count = Asset.objects.count()
        initial_link_count = Link.objects.count()

        response = self.client.post(url, data, format="multipart")

        self.assertEqual(response.status_code, 201)
        self.assertEqual(Asset.objects.count(), initial_asset_count + 1)
        self.assertEqual(Link.objects.count(), initial_link_count + 1)

        new_asset = Asset.objects.latest("created")
        self.assertEqual(new_asset.title, "My Linked Asset")
        self.assertTrue(Link.objects.filter(asset=new_asset, resource=self.resource).exists())

    def test_upload_asset_large_file(self):
        """
        Ensure that when a file upload exceeds the size limit,
        no asset is created and a 400 error is returned.
        """
        initial_asset_count = Asset.objects.count()

        uploadsizelimit_obj, _ = UploadSizeLimit.objects.update_or_create(
            slug="asset_upload_size",
            defaults={
                "description": "Max size for the uploaded assets file via API",
                "max_size": 5,
            },
        )
        limit = uploadsizelimit_obj.max_size
        big_file_content = b"a" * (int(limit * 1024 * 1024) + 1)
        big_file = self._create_dummy_file(filename="big_file.txt", content=big_file_content)

        url = reverse("base-resources-assets", kwargs={"pk": self.resource.pk})
        response = self.client.post(
            url,
            {
                "files": big_file,
                "title": "Big Asset",
            },
            format="multipart",
        )

        self.assertEqual(response.status_code, 400)

        self.assertEqual(Asset.objects.count(), initial_asset_count)

    def test_upload_wrong_document_type(self):
        """
        Tests that uploading a file with a disallowed extension fails.
        """
        self.client.force_login(self.test_user)
        url = reverse("base-resources-assets", kwargs={"pk": self.resource.pk})

        disallowed_file = self._create_dummy_file(filename="malicious_file.exe", content=b"malicious content")
        data = {"files": disallowed_file}

        response = self.client.post(url, data, format="multipart")

        self.assertEqual(response.status_code, 400)
        self.assertIn("The uploaded file type exe is not allowed.", str(response.content))

    def test_create_asset_no_add_permission(self):
        no_perm_user = get_user_model().objects.create_user(username="no_perm_user", password="password")
        self.client.force_login(no_perm_user)

        url = reverse("base-resources-assets", kwargs={"pk": self.resource.pk})
        file = self._create_dummy_file()
        data = {"files": file}

        response = self.client.post(url, data, format="multipart")

        self.assertEqual(response.status_code, 403)

    def test_create_asset_with_and_without_permission(self):
        """
        Test that a asset can be created with resource also with permission.
        """
        temp_user = get_user_model().objects.create_user(
            username="test_user_no_perms", email="noperms@example.com", password="testpass123"
        )

        url = reverse("base-resources-assets", kwargs={"pk": self.resource.pk})

        content_type = ContentType.objects.get_for_model(ResourceBase)
        # permission to post assets
        permission = Permission.objects.get(
            codename="add_resourcebase",
            content_type=content_type,
        )
        temp_user.user_permissions.add(permission)

        perm_spec = {
            "users": {
                temp_user.username: ["view_resourcebase"],
                self.admin_user.username: ["view_resourcebase", "change_resourcebase", "delete_resourcebase"],
            },
            "groups": {},
        }
        self.resource.set_permissions(perm_spec)

        self.client.force_login(temp_user)
        initial_asset_count = Asset.objects.count()

        file = self._create_dummy_file(filename="unauthorized_file.txt")
        data = {"files": file, "resource_id": self.resource.pk, "title": "Unauthorized Asset"}

        response = self.client.post(url, data, format="multipart")

        self.assertEqual(response.status_code, 403)

        self.assertEqual(Asset.objects.count(), initial_asset_count)

        # with admin user
        self.client.force_login(self.admin_user)
        admin_file = self._create_dummy_file(filename="admin_file.txt")
        admin_data = {"files": admin_file, "resource_id": self.resource.pk, "title": "Admin Asset"}

        admin_response = self.client.post(url, admin_data, format="multipart")
        self.assertEqual(admin_response.status_code, 201)
        self.assertEqual(Asset.objects.count(), initial_asset_count + 1)
        admin_asset = Asset.objects.latest("created")
        link = Link.objects.get(asset=admin_asset)
        self.assertEqual(link.asset.owner, self.admin_user)

        # with regular user and permission
        self.client.force_login(temp_user)
        perm_spec = {
            "users": {
                temp_user.username: ["change_resourcebase"],
                self.admin_user.username: ["view_resourcebase", "change_resourcebase", "delete_resourcebase"],
            },
            "groups": {},
        }
        self.resource.set_permissions(perm_spec)
        changed_asset_count = Asset.objects.count()
        changed_file = self._create_dummy_file(filename="new_change_file.txt")
        new_data = {"files": changed_file, "resource_id": self.resource.pk, "title": "Changed Asset"}
        changed_response = self.client.post(url, new_data, format="multipart")
        self.assertEqual(changed_response.status_code, 201)
        self.assertEqual(Asset.objects.count(), changed_asset_count + 1)
        new_asset = Asset.objects.latest("created")
        link = Link.objects.get(asset=new_asset)
        self.assertEqual(link.asset.owner, temp_user)


class AssetDeleteApiTests(GeoNodeBaseTestSupport):

    def setUp(self):
        super().setUp()
        self.admin_user = get_user_model().objects.get(username="admin")
        self.test_user = get_user_model().objects.create_user(
            username="test_user12", email="testuser@example.com", password="testpass123"
        )
        self.resource1 = ResourceBase.objects.create(
            title="Test Resource 1",
            owner=self.admin_user,
            uuid=str(uuid4()),
        )
        self.resource2 = ResourceBase.objects.create(
            title="Test Resource 2",
            owner=self.admin_user,
            uuid=str(uuid4()),
        )

        content_type = ContentType.objects.get_for_model(ResourceBase)
        permission = Permission.objects.get(
            codename="add_resourcebase",
            content_type=content_type,
        )
        self.test_user.user_permissions.add(permission)

        permission = Permission.objects.get(
            codename="delete_resourcebase",
            content_type=content_type,
        )
        self.test_user.user_permissions.add(permission)

        self.resource1.set_permissions(
            {"users": {self.test_user.username: ["view_resourcebase", "change_resourcebase", "delete_resourcebase"]}}
        )
        self.resource2.set_permissions(
            {"users": {self.test_user.username: ["view_resourcebase", "change_resourcebase", "delete_resourcebase"]}}
        )

        self.client.force_login(self.test_user)
        # Create asset 1 linked to resource 1
        self.asset1 = self._create_asset_via_api(self.resource1, "Asset 1")
        # Create asset 2 linked to resource 1 and 2
        self.asset2 = self._create_asset_via_api(self.resource1, "Asset 2")
        Link.objects.create(resource=self.resource2, asset=self.asset2)

    def _create_asset_via_api(self, resource, title):
        url = reverse("base-resources-assets", kwargs={"pk": resource.pk})
        file = SimpleUploadedFile("test_file.txt", b"test content", content_type="text/plain")
        data = {"files": file, "title": title}
        response = self.client.post(url, data, format="multipart")
        self.assertEqual(response.status_code, 201)
        return Asset.objects.get(pk=response.data["asset_id"])

    def test_delete_asset_as_owner(self):
        self.client.force_login(self.admin_user)
        url = reverse("base-resources-delete-asset", kwargs={"pk": self.resource1.pk, "asset_id": self.asset1.pk})
        response = self.client.delete(url)
        self.assertEqual(response.status_code, 200)
        self.assertFalse(Asset.objects.filter(pk=self.asset1.pk).exists())

    def test_delete_asset_with_permission(self):
        self.client.force_login(self.test_user)
        url = reverse("base-resources-delete-asset", kwargs={"pk": self.resource1.pk, "asset_id": self.asset1.pk})
        response = self.client.delete(url)
        self.assertEqual(response.status_code, 200)
        self.assertFalse(Asset.objects.filter(pk=self.asset1.pk).exists())

    def test_delete_asset_no_permission(self):
        no_perm_user = get_user_model().objects.create_user(username="no_perm", password="password")
        self.client.force_login(no_perm_user)
        url = reverse("base-resources-delete-asset", kwargs={"pk": self.resource1.pk, "asset_id": self.asset1.pk})
        response = self.client.delete(url)
        self.assertEqual(response.status_code, 403)
        self.assertTrue(Asset.objects.filter(pk=self.asset1.pk).exists())
